package com.hulk.rpc.server;

import com.hulk.rpc.common.EventLoopGroupHolder;
import com.hulk.rpc.config.common.HostPort;
import com.hulk.rpc.config.server.ServerConfig;
import com.hulk.rpc.filter.RestServerFilter;
import com.hulk.rpc.filter.RpcServerFilter;
import com.hulk.rpc.invoke.ServiceInvokeFactory;
import com.hulk.rpc.param.MethodParamClassResolver;
import com.hulk.rpc.registry.ServerMetaInfo;
import com.hulk.rpc.serialize.Serializer;
import com.hulk.rpc.config.server.ProtocolType;
import com.hulk.rpc.transport.server.Server;
import com.hulk.rpc.transport.server.rest.NettyRestServer;
import com.hulk.rpc.transport.server.rpc.NettyRpcServer;
import io.netty.channel.EventLoopGroup;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.Closeable;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;

/**
 * 顶层服务工厂，用于注册server、server方法调用
 *
 * @author xuxukang
 * @date 2019-03-09 17:46
 */
@Slf4j
@Data
public class HulkServer implements Closeable {

  /**
   * 用于处理rpc-server的多线程任务
   */
  private static final ForkJoinPool serverForkJoinPool = new ForkJoinPool(64);

  /**
   * 用于存储本实例启动的服务，可能会启动多个，比如对外的rpc服务，实例内的rest内置服务
   */
  private Map<HostPort, Server> serverMap = new HashMap<>();
  private Set<Integer> portSet = new HashSet<>();
  private boolean classIdRegistered = false;
  private volatile boolean isClosed = false;

  private final Serializer serializer;
  private final ServiceInvokeFactory serviceInvokeFactory;
  private final EventLoopGroup eventLoopGroup;
  private final ServerConfig serverConfig;
  private final CopyOnWriteArrayList<RpcServerFilter> rpcFilters = new CopyOnWriteArrayList<>();
  private final CopyOnWriteArrayList<RestServerFilter> restFilters = new CopyOnWriteArrayList<>();

  static {
    // 自动资源清理
    Runtime.getRuntime().addShutdownHook(
            new Thread(serverForkJoinPool::shutdownNow, "serverForkJoinPool-shutdown-thread"));
  }

  public HulkServer(ServerConfig serverConfig, ServiceInvokeFactory invokerFactory) {
    Objects.requireNonNull(serverConfig, "serverConfig");
    Objects.requireNonNull(invokerFactory, "invokerFactory");

    this.serverConfig = serverConfig;
    this.serviceInvokeFactory = invokerFactory;
    this.eventLoopGroup = EventLoopGroupHolder.get();
    this.serializer = serverConfig.getSerializer();

    MethodParamClassResolver classResolver = new MethodParamClassResolver(invokerFactory);
    this.serializer.setClassResolver(classResolver);
  }

  public HulkServer(String group, String app) {
    this(new ServerConfig(group, app), new ServiceInvokeFactory(group, app));
  }

  public void addFirst(RpcServerFilter filter) {
    rpcFilters.add(0, filter);
  }

  public void addLast(RpcServerFilter filter) {
    rpcFilters.add(filter);
  }

  public void addFirst(RestServerFilter filter) {
    restFilters.add(0, filter);
  }

  public void addLast(RestServerFilter filter) {
    restFilters.add(filter);
  }

  /**
   * 启动并注册服务(对外)
   *
   * @throws ExecutionException
   * @throws InterruptedException
   */
  public void startAndRegisterServer() throws ExecutionException, InterruptedException {
    if (serverConfig.getRegistryConfigs() == null || serverConfig.getRegistryConfigs().isEmpty()) {
      return;
    }

    serverForkJoinPool.submit(() -> {
      serverConfig.getRegistryConfigs().parallelStream().forEach(registerConfig -> {
        try {
          if (!serverMap.containsKey(registerConfig.getServerAddress())) {
            Server server = startServer(registerConfig.getProtocolType(), registerConfig.getServerAddress());

            portSet.add(registerConfig.getServerAddress().getPort());
            serverMap.put(registerConfig.getServerAddress(), server);
          }

          if (registerConfig.getServerRegistry() != null) {
            registerConfig.getServerRegistry().register(
                    new ServerMetaInfo().setGroup(serverConfig.getGroup())
                            .setApp(serverConfig.getApp())
                            .setHostPort(registerConfig.getServerAddress())
                            .setProtocolType(registerConfig.getProtocolType())
                            .setWeight(registerConfig.getServerWeight()));
          }

        } catch (Exception e) {
          if (log.isErrorEnabled()) {
            log.error("启动失败，" + registerConfig);
          }
        }
      });
    }).get();
  }

  private Server startServer(ProtocolType protocol, HostPort hostPort) throws InterruptedException {
    switch (protocol) {
      case rpc:
        return startRpcServer(hostPort);

      case rest:
        return startRestServer(hostPort);

      default:
        throw new UnsupportedOperationException();
    }
  }

  private NettyRpcServer startRpcServer(HostPort hostPort) throws InterruptedException {
    if (!classIdRegistered && serializer.isSupportedClassId()) {
      classIdRegistered = true;

      Map<String, Integer> classIdMap = serviceInvokeFactory.getClassIdMap();
      Map<Class<?>, Integer> classIds = new HashMap<>();

      classIdMap.forEach((className, id) -> {
        try {
          classIds.put(Class.forName(className), id);
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      });

      serializer.setClassIds(classIds);
      log.info("register Serializer.classIds: " + classIdMap);
    }

    NettyRpcServer nettyRpcServer = new NettyRpcServer(rpcFilters, hostPort, eventLoopGroup, serviceInvokeFactory, serializer);
    nettyRpcServer.start();

    return nettyRpcServer;
  }

  private NettyRestServer startRestServer(HostPort hostPort) throws InterruptedException {
    NettyRestServer nettyRestServer = new NettyRestServer(restFilters, hostPort, eventLoopGroup, serviceInvokeFactory,
            serverConfig.getJsonMapper());
    nettyRestServer.start();
    return nettyRestServer;
  }

  /**
   * 从注册中心删除服务
   */
  private void unregisterServer() throws ExecutionException, InterruptedException {
    if (serverConfig.getRegistryConfigs() == null || serverConfig.getRegistryConfigs().isEmpty()) {
      return;
    }

    serverForkJoinPool.submit(() -> {
      serverConfig.getRegistryConfigs().parallelStream().forEach(registerConfig -> {
        try {
          if (registerConfig.getServerRegistry() != null) {
            registerConfig.getServerRegistry().unregister(
                    new ServerMetaInfo().setApp(serverConfig.getApp())
                            .setGroup(serverConfig.getGroup())
                            .setHostPort(registerConfig.getServerAddress())
                            .setProtocolType(registerConfig.getProtocolType()));
          }
        } catch (Exception e) {
          if (log.isErrorEnabled()) {
            log.error("注销失败，" + registerConfig);
          }
        }
      });

      serverConfig.getRegistryConfigs().parallelStream().forEach(registerConfig -> {
        try {
          if (registerConfig.getServerRegistry() != null) {
            registerConfig.getServerRegistry().close();
          }
        } catch (Exception e) {
          if (log.isErrorEnabled()) {
            log.error("关闭失败，" + registerConfig);
          }
        }
      });
    }).get();
  }

  /**
   * 注册接口服务实际实现类
   *
   * @param classServiceMap 服务接口和实例的map
   */
  public void registerService(Map<Class<?>, Object> classServiceMap) {
    serviceInvokeFactory.register(classServiceMap);
  }

  @Override
  public void close() throws IOException {
    if (isClosed) {
      return;
    }

    isClosed = true;

    try {
      unregisterServer();
    } catch (Exception e) {
      // 不会出错的
    }

    serverMap.forEach((hostPort, server) -> {
      try {
        server.close();

        if (log.isInfoEnabled()) {
          log.info("成功关闭服务器: " + server.getClass().getName() + " " + hostPort);
        }
      } catch (Exception e) {
        if (log.isWarnEnabled()) {
          log.warn("关闭服务器出错: " + hostPort, e);
        }
      }
    });

    EventLoopGroupHolder.release(eventLoopGroup);
  }

}
